package org.example.hadoop.service;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.example.hadoop.domain.AuthorRatingScoreAvgBean;
import org.example.hadoop.domain.PublisherCountBean;
import org.example.hadoop.service.authorRatingScore.AuthorRatingScoreMapper;
import org.example.hadoop.service.authorRatingScore.AuthorRatingScoreReducer;
import org.example.hadoop.service.authorRatingScoreSort.AuthorRatingScoreSortMapper;
import org.example.hadoop.service.authorRatingScoreSort.AuthorRatingScoreSortReducer;
import org.example.hadoop.service.publishYearFrequency.PublishYearMapper;
import org.example.hadoop.service.publishYearFrequency.PublishYearReducer;
import org.example.hadoop.service.publisherSort.PublisherSortMapper;
import org.example.hadoop.service.publisherSort.PublisherSortReducer;
import org.example.system.SystemKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * @author wwy
 * @date 2020-11-10 19:49
 */
@Service
public class AuthorRatingScoreService {
    private static final Logger LOGGER = LoggerFactory.getLogger(AuthorRatingScoreService.class);
    @Value("${hadoop.inputPath}")
    private String inputPath;
    @Value("${hadoop.outputPath}")
    private String outputPath;

    /**
     * 每年发布的书籍
     */
    public void authorRatingScoreNumSort(){
        try {
            Configuration conf = new Configuration();
            Job job1 = Job.getInstance(conf,"authorRatingScoreNum");
            job1.setMapperClass(AuthorRatingScoreMapper.class);
            job1.setReducerClass(AuthorRatingScoreReducer.class);
            job1.setMapOutputKeyClass(Text.class);
            job1.setMapOutputValueClass(Text.class);
            job1.setOutputKeyClass(Text.class);
            job1.setOutputValueClass(Text.class);
            FileInputFormat.setInputPaths(job1, new Path(inputPath));
            FileOutputFormat.setOutputPath(job1, new Path(outputPath+"\\tem\\"+ SystemKey.AUTHOR_RATING_SCORE_AVG_FILE_NAME));
            //启动Job任务 ：向 yarn 集群提交这个 job

            Job job2 = Job.getInstance(conf,"authorRatingScoreNumSort");
            job2.setMapperClass(AuthorRatingScoreSortMapper.class);
            job2.setReducerClass(AuthorRatingScoreSortReducer.class);
            job2.setMapOutputKeyClass(AuthorRatingScoreAvgBean.class);
            job2.setMapOutputValueClass(Text.class);
            job2.setOutputKeyClass(Text.class);
            job2.setOutputValueClass(Text.class);
            FileInputFormat.setInputPaths(job2, new Path(outputPath+"\\tem\\"+SystemKey.AUTHOR_RATING_SCORE_AVG_FILE_NAME));
            FileOutputFormat.setOutputPath(job2, new Path(outputPath+"\\"+SystemKey.AUTHOR_RATING_SCORE_AVG_FILE_NAME));
            //启动Job任务 ：向 yarn 集群提交这个 job
            if (job1.waitForCompletion(true)) {
                job2.waitForCompletion(true);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}
